/**
 * Copyright ©2003~2019年5月29日 szkingdom
 * @author raohj
 * @date 2019年5月29日
 * @file RRocketMQConsumer.java
 * @comment : please input your comment to this file
 * @version : input version
 * @modify:
 * 		1、 2019年5月29日 由 raohj 新建文件
 */
package com.ccnu.demojava.mq.rocketmq.p2p;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author raohj
 * @date 2019年5月29日
 * @version 
 * @comment to add your comment
 */
public class RRocketMQConsumer_Filter {
	private static final Logger log = LoggerFactory.getLogger(RRocketMQConsumer_Filter.class);
	
	public static class RPushConmerRunnable implements Runnable{
		private String consumerName;
		
		/**
		 * @param consumerName
		 */
		public RPushConmerRunnable(String consumerName) {
			super();
			this.consumerName = consumerName;
		}

		@Override
		public void run() {
			DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group1");
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			consumer.setNamesrvAddr("localhost:9876");
			// 设置为集群模式
//			consumer.setMessageModel(MessageModel.CLUSTERING);
			try {
				consumer.subscribe("Topic_test1", MessageSelector.bySql("index between 0 and 5"));
				consumer.registerMessageListener(new MessageListenerConcurrently() {					
					@Override
					public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
						for (MessageExt msg : msgs) {
							log.info( "[" + consumerName + "]recv msg:" + new String(msg.getBody()));	
						}
						
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
				});
				
				consumer.start();
			} catch (MQClientException e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * @author raohj
	 * @date 2019年5月29日
	 * @comment 如果需要测试广播模式或者集群模式，则需要启动多个JVM
	 * @param 
	 * @return 
	 * @param args
	 */
	public static void main(String[] args) {
		// push consumer
		ExecutorService pushPool = Executors.newFixedThreadPool(10);
		pushPool.execute(new RPushConmerRunnable("consumer1"));
		pushPool.shutdown();
	}
}
